package com.sub.spark.core.rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;

/**
 * RDD行动算子方法
 */
public class RDDActionMethod {

    private static JavaSparkContext javaSparkContext;

    public static void main(String[] args) {

        //spark conf
        SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("sparkCore");
        //spark context
        javaSparkContext = new JavaSparkContext(conf);

        JavaRDD<Integer> parallelizeRdd = javaSparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5));

        //collect
        parallelizeRdd.collect().forEach(System.out::println);

        //count
        long count = parallelizeRdd.count();
        System.out.println("RDD action count():"+count);

        //first

        Integer first = parallelizeRdd.first();
        System.out.println("RDD action first():"+first);

        //take
        System.out.println("RDD action take(3):");
        parallelizeRdd.take(3).forEach(System.out::println);

        //countByKey
        JavaPairRDD<Integer, Integer> mapToPairRDD = parallelizeRdd.mapToPair(data -> new Tuple2<>(data, data));
        Map<Integer, Long> countByKey = mapToPairRDD.countByKey();
        System.out.println(countByKey);

        //foreach：分布式循环，分区内有序，分区间无序
        System.out.println("RDD action foreach()");
        parallelizeRdd.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });

        //foreachPartition: 对比foreach，分区，全量，效率高
        System.out.println("RDD action foreachPartition()");

        parallelizeRdd.foreachPartition(new VoidFunction<Iterator<Integer>>() {
            @Override
            public void call(Iterator<Integer> integerIterator) throws Exception {
                while (integerIterator.hasNext()){
                    System.out.println(integerIterator.next());
                }
            }
        });

        //扩展：RDD 行动算子（executor）中不能使用Java自己的函数式编程，如：parallelizeRdd.foreach(System.out::println); 无法序列化，Java中的函数式编程是对象模拟出来的，

        //保存出为对象
        parallelizeRdd.saveAsObjectFile("output1");


        //关闭jsc
        javaSparkContext.stop();

    }
}
